package com.dd.keel.core.framework.register;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.dd.keel.core.common.utils.FileUtil;
import com.dd.keel.core.framework.service.ServiceInstanceFactory;

/**
 * Register/Unregister service node
 * Zookeeper client library version must match Zookeeper server version.
 * @author Kevin.XU
 *
 */
public class RegisterUtil {

	final static private Logger LOGGER = LoggerFactory
			.getLogger(RegisterUtil.class);

	private ServiceInstanceFactory instanceFactory;
	private String zkAddr = null;

	private CuratorFramework zkClient;
	private String nodePath;

	public ServiceInstanceFactory getInstanceFactory() {
		return instanceFactory;
	}

	public void setInstanceFactory(ServiceInstanceFactory instanceFactory) {
		this.instanceFactory = instanceFactory;
	}

	public String getZkAddr() {
		return zkAddr;
	}

	public void setZkAddr(String zkAddr) {
		this.zkAddr = zkAddr;
	}

	/**
	 * register service node
	 * 
	 * @return
	 */
	public boolean registerServiceNode() {
		// read node config
		String nodeConfigFilePath = FileUtil
				.retrieveConfigFilepath("node.json");
		String nodeConfigContent = FileUtil.readTextFromFile(
				nodeConfigFilePath, "UTF-8");
		if (nodeConfigContent == null) {
			LOGGER.error("{} read failed", nodeConfigFilePath);
			return false;
		}
		JSONObject jSONObject = JSON.parseObject(nodeConfigContent);
		String host = jSONObject.getString("host");
		String port = jSONObject.getString("port");
		String group = jSONObject.getString("group");
		// read all service
		List<String[]> serviceConfigs = getInstanceFactory().getAllServices();
		if (serviceConfigs.size() == 0) {
			LOGGER.error("not found services");
			return false;
		}
		JSONObject root = new JSONObject();
		root.put("host", host);
		root.put("port", new Integer(port));
		root.put("group", group);
		JSONArray servicesNode = new JSONArray();
		for (int i = 0; i < serviceConfigs.size(); ++i) {
			String[] serviceConfig = serviceConfigs.get(i);
			JSONObject temp = new JSONObject();
			temp.put("service", serviceConfig[0]);
			temp.put("version", serviceConfig[1]);
			LOGGER.info("add service {}, {}", serviceConfig[0],
					serviceConfig[1]);
			servicesNode.add(temp);
		}
		root.put("services", servicesNode);
		// output
		String output = JSONObject.toJSONString(root, true);
		String serviceDefineFilePath = FileUtil
				.retrieveConfigFilepath("services.json");
		FileUtil.writeTextToFile(output, serviceDefineFilePath, "UTF-8");
		// register
		StringBuilder nodePathBuf = new StringBuilder();
		nodePathBuf.append('/').append(group).append("/providers/")
				.append(URLEncoder.encode(output));
		nodePath = nodePathBuf.toString();
		zkClient = CuratorFrameworkFactory.builder().connectString(zkAddr)
				.sessionTimeoutMs(10000).connectionTimeoutMs(10000)
				.retryPolicy(new RetryNTimes(3, 5000)).build();
		zkClient.getConnectionStateListenable().addListener(
				new ConnectionStateListener() {
					@Override
					public void stateChanged(CuratorFramework client,
							ConnectionState newState) {
						LOGGER.info("client state:" + newState.name());
						if (newState == ConnectionState.CONNECTED) {
							try {
								zkClient.create().creatingParentsIfNeeded()
										.withMode(CreateMode.EPHEMERAL)
										.withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
										.forPath(nodePath);
								LOGGER.info("create session successfully");
							} catch (InterruptedException ex) {
								LOGGER.error(ex.getMessage(), ex);
							} catch (Exception ex) {
								LOGGER.error(ex.getMessage(), ex);
							}
						} else if (newState == ConnectionState.LOST) {
							try {
								Stat stat = zkClient.checkExists().forPath(
										nodePath);
								if (stat == null) {
									LOGGER.info("recreate session");
									zkClient.create()
											.creatingParentsIfNeeded()
											.withMode(CreateMode.EPHEMERAL)
											.withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
											.forPath(nodePath);
									LOGGER.info("recreate session successfully");
								}
							} catch (Exception ex) {
								LOGGER.error(ex.getMessage(), ex);
							}
						}
					}
				});
		zkClient.start();
		return true;
	}

	/**
	 * unregister service node
	 */
	public void unregisterServiceNode() {
		zkClient.close();
	}

}
